package com.lagou.flink.work.p3;

import org.apache.flink.api.java.tuple.Tuple;

import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow;
import org.apache.flink.util.Collector;
import scala.Tuple2;



import java.text.SimpleDateFormat;

/**
 * 功能描述：
 *
 * @author : zhangyong
 * @since : 2021/5/18
 */
public class MyCountWindow implements WindowFunction<Tuple2<String, Integer>,
        String, Tuple, GlobalWindow> {

    @Override
    public void apply(Tuple tuple, GlobalWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss.SSS");
        int sum = 0;
        for (Tuple2<String, Integer> tuple2 : input){
            sum += tuple2._2;
        }
        long maxTimestamp = window.maxTimestamp();
        out.collect("key:" + tuple.getField(0) + " value: " + sum + "| maxTimeStamp :"
                + maxTimestamp + "," + format.format(maxTimestamp)
        );
    }
}
